package cn.itcast.sse;

import cn.itcast.avro.SseAvro;
import cn.itcast.util.DateTimeUtil;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.apache.commons.net.ftp.FTPReply;
import org.apache.flume.Context;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.AbstractSource;

import java.io.*;
import java.util.Date;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class SseQuotSource extends AbstractSource implements PollableSource, Configurable {
    private String host;
    private Integer port;
    private String userName;
    private String password;
    private String directory;
    private String fileName;
    private String localDirectory;
    private Integer delay;
    private Integer corePoolSize;
    private Integer maxPoolSize;
    private Long keepAliveTime;
    private Integer capacity;
    private ThreadPoolExecutor executor;

    /**
     * 业务处理
     */
    @Override
    public Status process() throws EventDeliveryException {

        //3.判断是否是交易时间段
        long time = new Date().getTime();
        if (time > DateTimeUtil.openTime && time < DateTimeUtil.closeTime) {

            //4.异步处理
            /**
             * 开发步骤：
             * 1.创建异步线程task
             * 2.下载行情文件
             * 3.解析并发送数据
             *   数据转换成avro
             *   数据序列化
             * 4.发送数据到channel
             */
            //4.异步处理
            executor.execute(new AsyncTask());

            //5.设置延时时间
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return Status.READY;
    }


    /**
     * 初始化方法
     */
    @Override
    public void configure(Context context) {
        host = context.getString("host");
        port = context.getInteger("port");
        userName = context.getString("userName");
        password = context.getString("password");
        directory = context.getString("directory");
        fileName = context.getString("fileName");
        localDirectory = context.getString("localDirectory");
        delay = context.getInteger("delay");
        corePoolSize = context.getInteger("corePoolSize");
        maxPoolSize = context.getInteger("maxPoolSize");
        keepAliveTime = context.getLong("keepAliveTime");
        capacity = context.getInteger("capacity");
        /**
         * corePoolSize：线程池核心线程数
         * maximumPoolSize：线程池最大线程数（当workQueue都放不下时，启动新线程，最大线程数）
         * keepAliveTime：超出corePoolSize数量的线程的保留时间
         * unit：keepAliveTime单
         * workQueue：阻塞队列，存放线程
         */
        executor = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(capacity));
    }

    private class AsyncTask implements Runnable {

        /**
         * 开发步骤：
         * 1.初始化ftp连接
         * （1）设置IP和Port
         * （2）设置登陆用户名和密码
         * (3) 设置编码格式
         * （4）判断是否连接成功（FTPReply）
         * 2.切换工作目录，设置被动模式
         * 3.获取工作目录的文件信息列表
         * 4.输出文件
         * 5.退出，返回成功状态
         */
        @Override
        public void run() {
            //下载ftp行情文件
            boolean bool = download(directory, localDirectory);


            try {
                if (bool == true) {
                    //1.读取文件获取行数据
                    InputStreamReader inputStreamReader = new InputStreamReader(new FileInputStream(localDirectory + "/" + fileName));
                    BufferedReader bufferedReader = new BufferedReader(inputStreamReader);
                    String line;
                    int i = 0;
                    while ((line = bufferedReader.readLine()) != null) {
                        //2.获取首行，判断市场行情状态
                        String[] arr = line.split("\\|");
                        if (i == 0) {
                            //市场行情状态
                            String status = arr[8];
                            if (status.startsWith("E")) {
                                break;
                            }
                        } else {
                            //3.数据转换成avro对象
                            SseAvro sseAvro = transform(arr);
                            //4.序列化操作
                            byte[] serialize = serialize(sseAvro);
                            //5.发送管道
                            getChannelProcessor().processEvent(EventBuilder.withBody(serialize));
                        }
                        i++;
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }


        }

        private byte[] serialize(SseAvro sseAvro) {
            final SpecificDatumWriter<Object> datumWriter = new SpecificDatumWriter<>(sseAvro.getSchema());

            final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
            try {

                final BinaryEncoder binaryEncoder = EncoderFactory.get().directBinaryEncoder(byteArrayOutputStream, null);
                datumWriter.write(sseAvro,binaryEncoder);
            } catch (IOException e) {
                e.printStackTrace();
            }
            return byteArrayOutputStream.toByteArray();
        }

        private SseAvro transform(String[] arr) {
            SseAvro sseAvro  = new SseAvro();
            sseAvro.setMdStreamID(arr[0].trim());
            sseAvro.setSecurityID(arr[1].trim());
            sseAvro.setSymbol(arr[2].trim());
            sseAvro.setTradeVolume(Long.valueOf(arr[3].trim()));
            sseAvro.setTotalValueTraded(Long.valueOf(arr[4].trim()));
            sseAvro.setPreClosePx(Double.valueOf(arr[5].trim()));
            sseAvro.setOpenPrice(Double.valueOf(arr[6].trim()));
            sseAvro.setHighPrice(Double.valueOf(arr[7].trim()));
            sseAvro.setLowPrice(Double.valueOf(arr[8].trim()));
            sseAvro.setTradePrice(Double.valueOf(arr[9].trim()));
            sseAvro.setClosePx(Double.valueOf(arr[10].trim()));
            sseAvro.setTradingPhaseCode("T01");
            sseAvro.setTimestamp(new Date().getTime());
            return sseAvro;
        }

        private boolean download(String directory, String localDirectory) {

            boolean flag = false;
            //1.初始化ftp连接
            FTPClient ftpClient = initFtpClient();

            try {
                //2.切换工作目录
                ftpClient.changeWorkingDirectory(directory);
                //被动模式，服务端开放端口，用于数据传输
                ftpClient.enterLocalPassiveMode();
                //禁用服务端参与的验证，如果不禁用服务端会获取主机IP与提交的host进行匹配，不一致时会报错
                ftpClient.setRemoteVerificationEnabled(false);
                //3.获取工作目录的文件信息列表
                FTPFile[] ftpFiles = ftpClient.listFiles();
                for (FTPFile ftpFile : ftpFiles) {
                    File file = new File(localDirectory + "/" + ftpFile.getName());
                    FileOutputStream fos = new FileOutputStream(file);
                    //4.文件输出
                    ftpClient.retrieveFile(ftpFile.getName(), fos);
                    fos.close();
                }
                ftpClient.logout();
                System.out.println("文件下载成功。。。。");
                //5.返回成功状态
                flag = true;
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                if (ftpClient.isConnected()) {
                    try {
                        ftpClient.disconnect();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
            return flag;
        }

        private FTPClient initFtpClient() {

            FTPClient ftpClient = new FTPClient();
            ftpClient.setControlEncoding("UTF-8");

            try {
                ftpClient.connect(host, port);
                ftpClient.login(userName, password);
                //FTP服务器检查连接是否成功,220表示连接成功，230登陆成功，221再见
                int replyCode = ftpClient.getReplyCode();
                //判断连接是否成功响应
                if (!FTPReply.isPositiveCompletion(replyCode)) {
                    System.out.println("connect ftp sever failed ");
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            return ftpClient;
        }
    }
}
